import java.net.*;
import java.io.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.*;
import java.nio.ByteBuffer;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.recipes.lock.WriteLock;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.ZooDefs.Ids;

public class AtomicChannel extends BroadcastChannel {
    private ReliableChannel rc;
    private ApplicationListener application;
    private WriteLock lock;
    private ZooKeeper zk;
    private int localCounter;
    private CopyOnWriteArrayList<ChatMessage> msgBuf;

    public AtomicChannel(ChatClientRecord rec, ApplicationListener l){
        application = l;
        //initialize the recieved message buffer
        msgBuf = new CopyOnWriteArrayList<ChatMessage>();

        //initialize the underlying reliable channel
        rc = new ReliableChannel(rec, this);

        String connectString = "sslab22:9000,sslab23:9000,sslab24:9000";

        try{
            //initialize zookeeper connection
            zk = new ZooKeeper(connectString, 3000, null);

            //create node for write lock if it doesn't exist
            if(null == zk.exists("/atomic", false)){
                byte[] zerobytes = ByteBuffer.allocate(4).putInt(0).array();
                List<ACL> acl = Ids.OPEN_ACL_UNSAFE;
                zk.create("/atomic", zerobytes, acl, CreateMode.PERSISTENT);
                localCounter = 0;
            }else{
                byte[] data = zk.getData("/atomic",false,null);
                localCounter = ByteBuffer.allocate(4).put(data).getInt(0);
            }

            lock = new WriteLock(zk, "/atomic", null);
        }catch(Exception e){
            System.err.println("ZooKeeper: "+e.getMessage());
            System.exit(1);
        }
    }

    //Add a client to this channel
    public boolean addMember(ChatClientRecord rec){
        return rc.addMember(rec);
    }

    //Remove a client from this channel
    public boolean removeMember(ChatClientRecord rec){
        return rc.removeMember(rec);
    }

    //Get the list of clients in this channel
    public CopyOnWriteArrayList<ChatClientRecord> getMembers(){
        return rc.getMembers();
    }

    public boolean sendMessage(ChatMessage m){
        try{
            //wait for lock
            lock.lock();
            while(!lock.isOwner()){ Thread.yield(); }

            //Get the current counter
            byte[] data = zk.getData("/atomic", false, null);
            int intdata = ByteBuffer.allocate(4).put(data).getInt(0);
            //Set the message's ID
            m.messID=intdata;
            intdata++;
            //Increment and set the counter's new value
            data = ByteBuffer.allocate(4).putInt(intdata).array();
            zk.setData("/atomic",data,-1);
            //Send message
            rc.sendMessage(m);

            //unlock
            lock.unlock();
        }catch(KeeperException e){
            System.out.println(e.getMessage());
        }catch(InterruptedException e){
            System.out.println(e.getMessage());
        }
        return false;
    }

    public void receive(ChatMessage m){
        msgBuf.add(m);
        boolean sent = true;
        //Loop over the buffer and deliver them in the correct order
        while(sent){
            sent = false;
            for(int i = 0; i < msgBuf.size(); i++){
                ChatMessage mess = msgBuf.get(i);
                if(mess.messID == localCounter){
                    application.receive(mess);
                    localCounter++;
                    msgBuf.remove(i);
                    sent = true;
                }
            }
        }
    }
}
